Pull request overview

This PR addresses a control-message future race in the computer-core message-sending pipeline (specifically QueuedMessageSender) that could surface as The origin future must be null, and also tightens computer-test integration behavior to fail faster instead of stalling for long BSP timeouts.

Changes:

• Refactors control-message handling in QueuedMessageSender so each START/FINISH message carries its own CompletableFuture, and the in-flight control future is cleared before completion.
• Adds unit regressions for consecutive control messages and for transport failures completing the control future exceptionally.
• Updates sender integration tests to apply shorter BSP wait timeouts and to use a bounded waitForServices() helper.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.

        public void sendFinishMessage(CompletableFuture<Void> future)
                                      throws TransportException {
            this.setControlFuture(future);
            try {
                this.client.finishSessionAsync().whenComplete((r, e) -> {

                                          .withRpcServerHost("127.0.0.1")
                                          .withRpcServerPort(8611)
                                          .withRpcServerPort(0)

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Blocking: yes. Summary: An existing review thread covers the sender-executor termination risk, and the new fail-fast service wait can race past lifecycle cleanup. Evidence: exact-head static inspection by six independent lanes; git diff --check passed; visible exact-head checks are green.

Blocking: yes. Summary: Synchronous runtime failures can still strand an in-flight control future, and the fail-fast cleanup/tests leave concurrency gaps. Evidence: six independent exact-head review lanes; targeted QueuedMessageSenderTest passed; git diff --check passed; visible exact-head checks are green.

Blocking: yes. Summary: Two control-message failure windows can still strand work or hide a connection failure, and initialization failures can bypass the new cleanup path; latest-head Computer CI is also failing. Evidence: exact-head static interleaving analysis; local Maven validation stopped before the target test because the available JDK cannot compile computer-k8s; GitHub Actions run 30206151982 reports three unit-test failures.

Blocking: yes. Summary: Two fail-fast cleanup paths can still extend CI stalls and hide the original service failure. Evidence: six independent exact-head review lanes; QueuedMessageSenderTest passed 7/7 on JDK 11; git diff --check passed; visible exact-head checks were green.

Blocking: yes. Summary: A synchronous data-path failure can still terminate the sole sender and strand queued control futures, while two new cleanup tests can leak non-daemon threads on an early assertion failure. Evidence: six independent exact-head review lanes; QueuedMessageSenderTest passed 7/7 on JDK 11; git diff --check passed; visible exact-head checks are green.

Blocking: yes. Summary: An uncaught master Error can leave the service future pending until the five-minute timeout; two previously reported sender and cleanup risks also remain covered by existing threads. Evidence: six independent exact-head review lanes, static control-flow and interleaving analysis, QueuedMessageSenderTest passed 9/9, and visible exact-head checks are green.

Blocking: yes. Summary: A data-send failure can still be forgotten before FINISH is registered, allowing the step to succeed after silently dropping a message; the new timeout regression also has a minor timing-flake window. Evidence: six independent exact-head review lanes; static MessageSendManager/QueuedMessageSender interleaving analysis; targeted QueuedMessageSenderTest passed 9/9; git diff --check passed; visible exact-head checks are green.

Blocking: yes. Summary: A control-completion race can silently discard transport failures, and service cleanup errors can still be hidden after successful execution. Evidence: six independent exact-head review lanes; targeted QueuedMessageSenderTest passed 12/12 on JDK 11; git diff --check passed; visible exact-head checks are green.

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated no new comments.

        int transoprtServerPort = 8998;

Blocking: yes. Summary: The new public control-message constructor can produce a future that is silently dropped by the existing queued-message send path. Evidence: six independent exact-head review lanes; QueuedMessageSenderTest passed 13/13 on JDK 11; git diff --check passed; visible exact-head checks are green.

Blocking: yes. Summary: The partial WorkerService initialization cleanup remains covered by an existing review thread, and the new transport-exception tests do not verify per-connection isolation. Evidence: six independent exact-head review lanes; QueuedMessageSenderTest passed 13/13 on JDK 11; git diff --check passed; visible exact-head checks are green.

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.

    public CompletableFuture<Void> send(int workerId, MessageType type)
                                        throws InterruptedException {
        WorkerChannel channel = this.channels[channelId(workerId)];
        CompletableFuture<Void> future = new CompletableFuture<>();
        if (!channel.setControlFuture(future)) {
            return future;
        }
        /*
         * Control message just need message type is enough,
         * partitionId = -1 and buffer = null represents a meaningless value
         */
        try {
            channel.queue.put(new QueuedMessage(-1, type, null, future));
        } catch (InterruptedException e) {
            channel.completeControlFuture(future, e);
            throw e;
        }
        return future;
    }

            ComputerException e = new ComputerException(
                                  "The origin future must be null");
            future.completeExceptionally(e);
            return false;

    private static boolean await(CountDownLatch latch) throws InterruptedException {
        return latch.await(1, TimeUnit.SECONDS);
    }

Pull request overview

Copilot reviewed 7 out of 7 changed files in this pull request and generated no new comments.

                                          .withRpcServerHost("127.0.0.1")
                                          .withRpcServerPort(8611)
                                        .withRpcServerPort(0)